#pragma once

#include "../common/Net.hpp"
#include "../common/Message.hpp"
#include "Requester.hpp"


namespace rpc
{
    namespace client
    {
        class TopicManager
        {
        public:
            using ptr = std::shared_ptr<TopicManager>;
            // <topic, msg>
            using SubsCallback = std::function<void(const std::string&, const std::string&)>;

            TopicManager(const Requester::ptr &requester)
                : _requester(requester)
            {}

            // 处理接收到的请求消息
            void onPublish(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                // 1. 判断主题操作类型是否正确
                auto operation = msg->topicOption();
                if (operation != TOPIC_OP_TYPE::TOPIC_PUBLISH)
                {
                    ELOG("topic operation type error");
                    return;
                }
                // 2. 取出主题类型和消息
                auto topic_key = msg->topicTheme();
                auto topic_msg = msg->topicMsg();

                // 3. 获取回调函数
                auto callback = getSubsCallback(topic_key);
                if (!callback)
                {
                    ELOG("not found [%s] corresponding callback", topic_key.c_str());
                    return;
                }

                callback(topic_key, topic_msg);
            }

            bool topicCreate(const BaseConnection::ptr &conn, const std::string &topic_key)
            {
                return commRequestDeal(conn, topic_key, TOPIC_OP_TYPE::TOPIC_CREATE);
            }

            bool topicRemove(const BaseConnection::ptr &conn, const std::string &topic_key)
            {
                return commRequestDeal(conn, topic_key, TOPIC_OP_TYPE::TOPIC_REMOVE);
            }

            bool topicSubscrive(const BaseConnection::ptr &conn, const std::string &topic_key, const SubsCallback &cb)
            {
                addSubsCallback(topic_key, cb);
                bool ret = commRequestDeal(conn, topic_key, TOPIC_OP_TYPE::TOPIC_SUBSCRIBE);
                if (ret == false)
                {
                    delSubsCallback(topic_key);
                    return false;
                }
                return true;
            }

            bool topicUnsubscribe(const BaseConnection::ptr &conn, const std::string &topic_key)
            {
                delSubsCallback(topic_key);
                return commRequestDeal(conn, topic_key, TOPIC_OP_TYPE::TOPIC_CANCEL);
            }

            bool topicPublish(const BaseConnection::ptr &conn, const std::string topic_key, const std::string msg)
            {
                return commRequestDeal(conn, topic_key, TOPIC_OP_TYPE::TOPIC_PUBLISH, msg);
            }
        
        private:
            SubsCallback getSubsCallback(const std::string &topic_key)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _topic_callback.find(topic_key);
                if (it == _topic_callback.end())
                    return SubsCallback();
                
                return it->second;
            }

            void addSubsCallback(const std::string &topic_key, const SubsCallback &cb)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _topic_callback.insert(std::make_pair(topic_key, cb));
            }

            void delSubsCallback(const std::string &topic_key)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _topic_callback.erase(topic_key);
            }

            bool commRequestDeal(const BaseConnection::ptr &conn, const std::string &topic_key, TOPIC_OP_TYPE type,
                const std::string &msg = "")
            {
                // 1. 创建请求
                auto msg_req = MessageFactory::create<TopicRequest>();
                msg_req->setId(UUID::uuid());
                msg_req->setTopicTheme(topic_key);
                msg_req->setMsgType(MSG_TYPE::REQ_TOPIC);
                msg_req->setTopicOption(type);

                if (type == TOPIC_OP_TYPE::TOPIC_PUBLISH)
                {
                    msg_req->setTopicMsg(msg);
                }

                // 2. 发送请求
                BaseMessage::ptr base_rsp;
                bool ret = _requester->send(conn, msg_req, base_rsp);
                if (ret == false)
                {
                    ELOG("topic request send error");
                    return false;
                }

                // 3. 处理响应
                auto msg_rsp = std::dynamic_pointer_cast<TopicResponse>(base_rsp);
                if (msg_rsp.get() == nullptr)
                {
                    ELOG("topic response type down error");
                    return false;
                }
                if (msg_rsp->rcode() != RCODE::RCODE_OK)
                {
                    ELOG("topic request error");
                    return false;
                }
                return true;
            }

        private:
            std::mutex _mutex;
            // <topic_key, callback>
            std::unordered_map<std::string, SubsCallback> _topic_callback;
            Requester::ptr _requester;
        };
    }
}